package org.apache.rocketmq.proxy.processor;

import io.netty.channel.Channel;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
import org.apache.rocketmq.broker.client.ProducerChangeListener;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.service.ServiceManager;
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;

import java.util.Set;

public class ClientProcessor extends AbstractProcessor {

    public ClientProcessor(MessagingProcessor messagingProcessor, ServiceManager serviceManager) {
        super(messagingProcessor, serviceManager);
    }

    public void registerProducer(ProxyContext ctx, String producerGroup, ClientChannelInfo clientChannelInfo) {
        this.serviceManager.getProducerManager().registerProducer(producerGroup, clientChannelInfo);
    }

    public void unRegisterProducer(ProxyContext ctx, String producerGroup, ClientChannelInfo clientChannelInfo) {
        this.serviceManager.getProducerManager().unregisterProducer(producerGroup, clientChannelInfo);
    }

    public Channel findProducerChannel(ProxyContext ctx, String producerGroup, String clientId) {
        return this.serviceManager.getProducerManager().findChannel(clientId);
    }

    public void registerProducerChangeListener(ProducerChangeListener listener) {
        this.serviceManager.getProducerManager().appendProducerChangeListener(listener);
    }

    public void registerConsumer(ProxyContext ctx, String consumerGroup, ClientChannelInfo clientChannelInfo, ConsumeType consumeType, MessageModel messageModel, ConsumeFromWhere consumeFromWhere, Set<SubscriptionData> subList, boolean updateSubscription) {
        this.serviceManager.getConsumerManager().registerConsumer(consumerGroup, clientChannelInfo, consumeType, messageModel, consumeFromWhere, subList, false, updateSubscription);
    }

    public ClientChannelInfo findConsumerChannel(ProxyContext ctx, String consumerGroup, Channel channel) {
        return this.serviceManager.getConsumerManager().findChannel(consumerGroup, channel);
    }

    public void unRegisterConsumer(ProxyContext ctx, String consumerGroup, ClientChannelInfo clientChannelInfo) {
        this.serviceManager.getConsumerManager().unregisterConsumer(consumerGroup, clientChannelInfo, false);
    }

    public void doChannelCloseEvent(String remoteAddr, Channel channel) {
        this.serviceManager.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
        this.serviceManager.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
    }

    public void registerConsumerIdsChangeListener(ConsumerIdsChangeListener listener) {
        this.serviceManager.getConsumerManager().appendConsumerIdsChangeListener(listener);
    }

    public ConsumerGroupInfo getConsumerGroupInfo(ProxyContext ctx, String consumerGroup) {
        return this.serviceManager.getConsumerManager().getConsumerGroupInfo(consumerGroup);
    }
}
